package com.lion.seaman.worker.flows;

import com.lion.seaman.common.tools.StringTools;
import com.lion.seaman.common.type.JobFlowStatus;
import org.apache.log4j.Logger;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

public class JobFlowGroup extends JobFlow {
    protected Logger log = Logger.getLogger(JobFlowGroup.class);
    protected Semaphore semaphore;
    protected Map<String, String> flows = new HashMap<>();
    protected ExecutorService executor;

    private JobFlowGroup(String flowId, int semaphore) {
        super(flowId, "");
        Assert.hasLength(flowId, "flowId path must not be null");
        Assert.isTrue(semaphore > 0, "semaphore must be greater than 0");
        this.semaphore = new Semaphore(semaphore);
    }

    @Override
    public void run() {
        try {
            if (!flows.isEmpty()) {
                CountDownLatch cdl = new CountDownLatch(flows.size());
                executor = Executors.newFixedThreadPool(flows.size());
                List<JobFlow> jobFlows = new ArrayList<>();
                for (Map.Entry<String, String> entry : flows.entrySet()) {
                    semaphore.acquire();
                    String flowId = entry.getKey();
                    String scriptPath = entry.getKey();
                    JobFlow jobFlow = new JobFlow(flowId, scriptPath, cdl, semaphore);
                    jobFlows.add(jobFlow);
                    executor.execute(jobFlow);
                }
                cdl.await();
                for (JobFlow jobFlow : jobFlows) {
                    if (jobFlow.getJobFlowStatus() == JobFlowStatus.ERROR) {
                        setJobFlowStatus(JobFlowStatus.ERROR);
                        return;
                    }
                }
                setJobFlowStatus(JobFlowStatus.SUCCESS);
            }
        } catch (Exception e) {
            setJobFlowStatus(JobFlowStatus.ERROR);
            e.printStackTrace();
            log.error(StringTools.format("flow {0} has error,{1}", getFlowId(), e.getMessage()), e);
        } finally {
            log.info(StringTools.format("flowGroup {0} is out of,status:{1}", flowId, jobFlowStatus));
        }
    }

    public JobFlowGroup add(String flowId, String scriptPath) {
        Assert.hasLength(flowId, "flowId path must not be null");
        Assert.hasLength(scriptPath, "script path must not be null");
        flows.put(flowId, scriptPath);
        return this;
    }

    public static JobFlowGroup create(String flowId, int semaphore) {
        return new JobFlowGroup(flowId, semaphore);
    }


    public Map<String, String> getFlows() {
        return flows;
    }

    public void setFlows(Map<String, String> flows) {
        this.flows = flows;
    }
}
